package org.lsy.filter;

import javafx.util.converter.LongStringConverter;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.lang.reflect.Array;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import java.util.stream.Stream;

/**
 * @author liusy95
 * @version 1.0
 * @date 2020/7/2
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {

    private static final String FALSE = "false";

    Map<String, List<Record>> records = new ConcurrentHashMap<>();

    public TPMonitorFilter () {
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);
        scheduledThreadPool.scheduleAtFixedRate(()->{
            System.out.println("start");
            records.entrySet().forEach(entry -> printTP(entry.getKey(), entry.getValue()));
            System.out.println("end");
        },5,5, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        String method = invocation.getMethodName();
        String async = RpcContext.getContext().getAttachment(Constants.ASYNC_KEY);
        try {
            return invoker.invoke(invocation);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        } finally {
            if (async == null || FALSE.equals(async)) {
                Long responseTime = System.currentTimeMillis() - startTime;
                if (!records.containsKey(method)) {
                    List<Record> list = new ArrayList(2000);
                    records.put(method, list);
                }
                records.get(method).add(new Record(System.currentTimeMillis(), responseTime));
            }
        }
    }

    private void printTP(String methodName, List<Record> responseTimes) {
        List<Record> collect = responseTimes.stream()
                .filter(record -> System.currentTimeMillis() - record.getTimestamp() <= 60000L)
                .sorted(Comparator.comparing(Record::getResponseTime))
                .collect(Collectors.toList());
        records.put(methodName, collect);
        int size = collect.size();
        //打印tp90
        int index = (int) (size * 0.9);
        System.out.println(methodName + " - TP90: " + collect.get(index).getResponseTime());
        //打印tp99
        index = (int) (size * 0.99);
        System.out.println(methodName + " - TP99: " + collect.get(index).getResponseTime());
    }

    class Record {
        private long timestamp;
        private long responseTime;

        public Record(long timestamp, long responseTime) {
            this.timestamp = timestamp;
            this.responseTime = responseTime;
        }

        public long getTimestamp() {
            return timestamp;
        }

        public void setTimestamp(long timestamp) {
            this.timestamp = timestamp;
        }

        public long getResponseTime() {
            return responseTime;
        }

        public void setResponseTime(long responseTime) {
            this.responseTime = responseTime;
        }
    }
}
